package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class DwdTradeCancelDetail {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        //1.3 设置状态的TTL  生产环境设置为最大乱序程度
//        tenv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));


        //todo 读取kafka订单预处理的表
        tenv.executeSql("create table dwd_order_pre ( " +
                "    `id` string, " +
                "    `order_id` string, " +
                "    `sku_id` string, " +
                "    `sku_name` string, " +
                "    `order_price` string, " +
                "    `sku_num` string, " +
                "    `create_time` string, " +
                "    `source_type_id` string, " +
                "    `source_type_name` string, " +
                "    `source_id` string, " +
                "    `split_total_amount` string, " +
                "    `split_activity_amount` string, " +
                "    `split_coupon_amount` string, " +
                "    `consignee` string, " +
                "    `consignee_tel` string, " +
                "    `total_amount` string, " +
                "    `order_status` string, " +
                "    `user_id` string, " +
                "    `payment_way` string, " +
                "    `delivery_address` string, " +
                "    `order_comment` string, " +
                "    `out_trade_no` string, " +
                "    `trade_body` string, " +
                "    `operate_time` string, " +
                "    `expire_time` string, " +
                "    `process_status` string, " +
                "    `tracking_no` string, " +
                "    `parent_order_id` string, " +
                "    `province_id` string, " +
                "    `activity_reduce_amount` string, " +
                "    `coupon_reduce_amount` string, " +
                "    `original_total_amount` string, " +
                "    `feight_fee` string, " +
                "    `feight_fee_reduce` string, " +
                "    `refundable_time` string, " +
                "    `order_detail_activity_id` string, " +
                "    `activity_id` string, " +
                "    `activity_rule_id` string, " +
                "    `order_detail_coupon_id` string, " +
                "    `coupon_id` string, " +
                "    `coupon_use_id` string, " +
                "    `type` string, " +
                "    `old` map<string,string>, " +
                "    row_op_ts TIMESTAMP_LTZ(3) "+
                ") " + MyKafkaUtil.getKafkaDDL("dwd_trade_order_pre_process","order_detail_2209a"));


        //todo 过滤出取消订单数据
        Table filterTable = tenv.sqlQuery("select " +
                "id, " +
                "order_id, " +
                "sku_id, " +
                "sku_name, " +
                "order_price, " +
                "sku_num, " +
                "operate_time cancel_time, " +
                "source_type_id, " +
                "source_type_name, " +
                "source_id, " +
                "split_total_amount, " +
                "split_activity_amount, " +
                "split_coupon_amount, " +
                "user_id, " +
                "province_id, " +
                "activity_id, " +
                "activity_rule_id, " +
                "coupon_id " +
                "from dwd_order_pre " +
                "where `type` = 'update' " +
                "and `old`['order_status'] is not null " +
                "and order_status = '1003' ");

        tenv.createTemporaryView("filter_table",filterTable);
//        tenv.toAppendStream(filterTable, Row.class).print(">>>>>>>>");

        //todo 创建kafka 取消订单表
        tenv.executeSql("create table dwd_trade_cancel_detail (" +
                "id string, " +
                "order_id string, " +
                "sku_id string, " +
                "sku_name string, " +
                "order_price string, " +
                "sku_num string, " +
                "cancel_time string, " +
                "source_type_id string, " +
                "source_type_name string, " +
                "source_id string, " +
                "split_total_amount string, " +
                "split_activity_amount string, " +
                "split_coupon_amount string, " +
                "user_id string, " +
                "province_id string, " +
                "activity_id string, " +
                "activity_rule_id string, " +
                "coupon_id string " +
                ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_trade_cancel_detail"));

        tenv.executeSql("insert into dwd_trade_cancel_detail select * from filter_table");
//        tenv.sqlQuery("select * from dwd_trade_cancel_detail").execute().print();


//        env.execute();
    }
}
